Agents: steering, a message joins the turn being answered at its next step by igalshilman · Pull Request #301 · restatedev/sdk-python

igalshilman · 2026-11-09T08:18:26Z

Summary

Steering lets a person add to a request while the agent is still working on it ("use Celsius", "skip Friday"). A message sent to the new steer endpoint joins the turn being answered at its next step, so the model reads it in its very next request. If the turn has no step left, the message is answered as the next turn instead. Either way it is answered exactly once, through crashes, retries, stops and kills.

chat is unchanged, and a turn that nobody steers makes no extra calls.

Turn semantics, before and after

	Before	With steering
A turn is	one prompt, and its answer	one prompt, the messages steered to it that it took, and one answer to all of them
A message sent while a turn runs	waits in the key's queue, and is the next turn	sent to `chat`: the same. Sent to `steer`: the running turn takes it at its next step, or else it is the next turn
Order	turns answer messages in the order they came	unchanged: a steered message keeps its place in the queue, unless the running turn takes it first
A turn stopped, failed or killed	its prompt is not answered	the same for the messages it took; those it did not take are answered as turns of their own

A step is the point where the model's reply has a result for each of its tool calls, just before the model's next request. It is the only place a turn takes steered messages, so the model never gets one between a tool call and its result.

What happens to a steered message

The message is sent	It is answered
while the turn has a step ahead (a tool call is running, say)	by the running turn, which takes it at that step
while the model writes the turn's final answer	as the next turn, before any message sent after it
when no turn is running	as a turn of its own, which it starts
to an agent whose turns take none, or past a turn's 16	as a turn of its own, after the running turn
to a turn that then stops, fails, is refused or killed	if the turn took it: not at all, as it ended with the turn; otherwise as a turn of its own

The endpoint answers {id, turn}: the message's id, and the turn that holds it, or null when it will be a turn of its own. Clients should send it with an idempotency key, so a retried request is the same message.

How it works

The steer view passes the message to the agent's companion (signals).
The companion always queues a private steered invocation for it on the key (idempotency key steer:<id>), so the message keeps its place in line. If the running turn takes steered messages, the companion also holds it in an inbox for that turn.
The turn learns of it from publish, which it already calls at each tool call's start and end: the response now lists the messages waiting. At its next step, the framework integration calls take_steers. That stores each message once, by its id, as a user message of the turn, made by the framework's own Items (steered, by default a prompt's item), and returns those items, which the integration puts into the request about to go out. The same transaction sends the companion taken, through its outbox.
steered runs once the turn has ended. If a turn stored the message, it does nothing; otherwise it answers it as an ordinary turn.

Exactly once comes from two unique keys: one steered invocation per message id (Restate's idempotency), and at most one stored copy per id (a unique column of the new restate_agent_steers table), which steered checks.

The conversation owns which messages were steered: it stores them as the framework marks them, so history(), a retried attempt and the next turns hand them back as they were, and no integration looks them up again.

Taken at the next step

sequenceDiagram
  autonumber
  participant C as Client
  participant V as Agent (steer view)
  participant S as Companion
  participant A as Agent (chat)
  participant P as Agent (steered)
  participant DB as Key's database
  C->>A: chat/send("Plan my week")
  A->>S: start(turn, steers)
  A->>DB: the model's reply, a tool call
  C->>V: steer("Skip Friday")
  V->>S: steer(id, text)
  S-)P: steered(id, text): queued now, behind the turn
  Note over S: inbox: the message, for this turn
  A->>S: publish (the tool call ends)
  S-->>A: revision, steering: [the message]
  A->>DB: the tool's result
  A->>DB: the message, a step of the turn, and taken(id) in the transaction's outbox
  DB--)S: taken(id), once the transaction commits
  A->>A: the model's next request has it
  A->>DB: the answer
  A->>S: close(turn)
  Note over P: runs once the turn ended: the message is stored, so it does nothing

Too late: the next turn

The model is already writing its final answer, so no step follows. The turn ends without the message, and the message's own turn answers it next, before a chat sent after it.

sequenceDiagram
  autonumber
  participant C as Client
  participant S as Companion
  participant A as Agent (chat)
  participant P as Agent (steered)
  participant DB as Key's database
  C->>A: chat/send("Plan my week")
  Note over A: the tools are done: the model writes the final answer
  C->>S: steer("Skip Friday"), through the steer view
  S-)P: steered(id, text): queued behind the turn
  Note over S: inbox: the message, for this turn
  C->>A: chat/send("And next week?"): queued behind steered
  A->>DB: the answer
  A->>S: close(turn): the message it did not take is forgotten here
  P->>DB: stored by a turn? no
  P->>DB: a turn of its own: "Skip Friday", and its answer
  Note over A: then "And next week?" is answered

The turn is stopped

A message the turn took is its own, like its prompt, so it is stopped with it. A message the turn had not taken yet is answered as a turn of its own afterwards. A turn that is stopped while it waits to store a message takes none.

sequenceDiagram
  autonumber
  participant C as Client
  participant S as Companion
  participant A as Agent (chat)
  participant P as Agent (steered)
  participant DB as Key's database
  C->>S: steer("Use Celsius"), through the steer view
  S-)P: steered(1): queued behind the turn
  A->>DB: at a step: "Use Celsius", a step of the turn, and taken(1)
  C->>S: steer("And the rain")
  S-)P: steered(2): queued behind steered(1)
  C->>A: stop
  A->>DB: the turn ends cancelled, and "Use Celsius" with it
  A->>S: close(turn): message 2, not taken, is forgotten here
  P->>DB: steered(1): stored by a turn? yes: nothing to do
  P->>DB: steered(2): stored? no: a turn of its own, "And the rain", and its answer

A crash after the message is stored

The message and taken commit together, so a retried attempt finds the message in the turn, gives it to the model in place, and never takes it again.

sequenceDiagram
  autonumber
  participant A1 as Attempt 1
  participant A2 as Attempt 2
  participant DB as Key's database
  participant S as Companion
  participant P as Agent (steered)
  A1->>DB: at a step: the message and taken(id), in one transaction
  Note over A1: the process dies before the model's request
  DB--)S: taken(id), from the outbox: sent again after a crash, with the same key
  A2->>DB: load: the turn has the message, after the tool's result
  Note over A2: the model's next request has it, and it is never taken again
  A2->>DB: the answer
  A2->>S: close(turn)
  P->>DB: stored by a turn? yes: nothing to do

What the model and clients see

The model. In the running turn, a steered message sits where it was taken: after the tool results, before the next request. Later turns see an earlier turn as its prompt, the messages steered to it, then its answer, in history(), the budget and summaries.
The turn's ask. latest_ask skips steered messages, so the output re-ask, ADK's resumed invocation and a graph's run behave as before.
Clients. Both messages views show steered messages under their own ids, and status.steering lists those still waiting. AgentClient.steer and /steer in the terminal chat send one.

Frameworks

Each integration takes the messages only as a model request goes out, and puts the items take_steers returns into that request:

OpenAI: the input filter takes them at each step. The SDK builds each request's input afresh, so the run (_Steers) splices them back into every later request, after the item each came after. A run also takes the messages waiting as it starts. Input guardrails don't run again. An agent used as a tool (Agent.as_tool) runs with the run's run_config, our filter with it: its requests get none of the messages.
Pydantic AI: an innermost _Steering capability, wrapped by _Snapshots, hands them to Pydantic AI's own RunContext.enqueue(priority='asap'). A snapshot stored by the same request covers only what came before them.
ADK: _Restate.before_model_callback, for an LlmAgent root, adds them to the request and to the session. Each has an invocation id of its own (_Items.steered), so a retry still resumes the prompt's invocation. The agent's own before_model_callback runs afterwards on the request with them, so a response it gives in place of the model's answers them too.
LangGraph's built-in agent: _Items.steered gives each the id steer:<id>, which is stored with it, and the model node keeps them in the run's state. A trimmed request never starts at one.

Limits

16 per turn. A turn takes at most 16 steered messages in all (STEERS); the companion counts the ones it took until the turn ends. Past that, each is a turn of its own.
Only at a step. A tool that waits a day for an approval holds the message until it returns; stop the turn to change course sooner.
Only as a request goes out. A message that no model request follows is a turn of its own. ADK, for one, asks the model nothing more after some tools (skip_summarization, set_model_response).
Agents that take none, so each steered message is a turn of its own:
- a LangGraph graph= agent;
- an ADK agent with a workflow root;
- a background agent;
- an OpenAI run after a handoff whose input_filter drops the handoff's own output (see below).

Traffic

Measured on a real server with scripts/steering_traffic.py, for a turn of one tool call:

	Journal entries	Companion calls
A turn nobody steers, on `main` and on this branch	47	5 (`start`, 3 × `publish`, `close`)
A turn that takes a steered message	57	6 (+ `taken`, from the outbox)
The message itself	`steer` view: 5, companion `steer`: 13, placeholder that finds it taken: 3

When nobody steers, a turn makes no extra calls. Inside the companion, publish does one more state read, of the empty inbox.

Compatibility

Wire format: additive only (Started.steers, publish's steering, Status.steering). A TypeScript companion has no steer handler, so the steer view fails against it, and turns take nothing from it. The port is step 8 of docs/agents/porting.md.
Reserved names: steer and steered are now built-in handler names. A user handler with either name is refused.
Schema: one appended migration, a new table.

Testing

45 new tests, and 13 durability mutants in scripts/resurrect.py (all caught).

In-process (tests/agents_steering.py, 11 tests): stored once, after a crash too; nothing taken while a call is open, outside a chat turn, or once the turn stopped; earlier turns as prompt, steered messages, answer; the budget counts them at once; the output check still asks again once; a drop frees the message; the summary view and cursors; the migration.
Companion, real server (tests/agents_signals.py, 3 new): the inbox and the placeholder; publish and taken; a closed turn forgets its messages; clear; the cap, which counts messages already taken.
Agent with a scripted framework, real server (tests/agents_steering_server.py, 5 tests): taken at the next step; too late, answered next in order; no turn running; an agent that takes none; a killed turn.
Per framework (26 tests): each checks the message reaches the next request once, and that a crash after it was stored retries with it once. Also:
- OpenAI (9): a message during the answer; its place in every later request, also after a handoff filter; before the first request; no input guardrail runs again; an agent used as a tool gets none of them, and the turn stores none of its items.
- Pydantic AI (6): a message during the answer; a snapshot in the same request; feedback and a message through a retry; prompt without snapshots.
- ADK (6): a message during the answer; a workflow root; a step after which ADK asks nothing; a fresh invocation that a retry resumes past.
- LangGraph (5): the message stays after the next round, and keeps its id through a retry; a request never starts at it; a graph= agent takes none.
Full suite, with main merged in: 607 passed. ruff, pyright (strict) and mypy are clean.

Found on the way

Fixed here:
- Pydantic AI stored a RetryFeedbackPart request again whenever a run merged or resumed it, on a retry and then on every later turn. _Restate now recognizes feedback by its parts.
- An OpenAI agent used as a tool (Agent.as_tool) runs with the outer run's run_config, so the turn's input filter stored its tool outputs, without their calls, which broke the next turn's request. A request made while one of the run's tool calls runs is now the nested run's, and gets the user's own filter only.
Not fixed here, documented in frameworks.md:
- An OpenAI handoff whose input_filter drops the handoff's own output (remove_all_tools) never stores that output. The next turns' requests then carry a function call with no output, which OpenAI's APIs refuse. Steering takes nothing after such a handoff.
- What other Pydantic AI capabilities enqueue (the harness's system reminders, loop detection, and so on) reaches the model but is never stored, so a retry and the next turns lack it.

Docs

turns.md is the reference: a turn is now defined as its prompt, the messages steered to it, and their answer, through its life, its end, retries and stops, and Steering a turn has the cases above. Also frameworks.md (a row, Items.steered, and notes per framework), conversation.md, README.md, learnings.md (D6 to D10), porting.md, access-patterns.md and testing.md.

🤖 Generated with Claude Code

igalshilman · 2026-11-09T10:34:42Z

@pullfrog review

pullfrog

ℹ️ No issues in the steering logic. One stale docstring reference, inline.

Reviewed changes

I reviewed the whole steering feature: the companion's inbox and placeholder protocol, how the conversation stores steered messages and reads them back, and each framework integration that takes them at a step.

Companion protocol (_signals.py): steer always sends the steered placeholder with the idempotency key steer:<id>. It holds the message in a per-turn inbox capped at STEERS, and taken marks it taken. publish returns steering. _closed/reset forget the inbox. I traced the races (steer around a turn's close, taken arriving after the close, a turn retried with the same invocation id). Each message ends up answered exactly once.
Storage (_conversation.py): Turn.steer mirrors _insert. It stores each message once, using the UNIQUE id of the new restate_agent_steers table, and sends taken from the same transaction's outbox. It is shielded, and checks going() again once it holds the ledger lock. AttemptEnded also sets stop.stopped, so a store waiting on the lock when an attempt ends takes nothing.
Reading back (_conversation.py, _messages.py): earlier turns carry steers, which feed history(), the budget, summaries and the summary view. The page reads cover the right ranges, and last_of derives seq from the tables, so the cursor moves forward for steered rows.
latest_ask skips steered messages, so the output re-ask, ADK's resumed invocation and graph runs work as before.
Framework integrations:
- OpenAI: the input filter splices the messages back into each later request, and the run also takes them as it starts.
- Pydantic AI: an innermost _Steering wrapped by _Snapshots calls RunContext.enqueue(priority='asap'). I checked this signature and its semantics against the installed pydantic_ai 2.66.
- ADK: _Restate is now added per run, and before_model_callback adds the messages to the request and to the session.
- LangGraph: the model node marks the messages with steer: ids, and _latest never starts a trimmed request at one.
Other drop callers: the OpenAI drop call sites (_resume, the output-guardrail re-ask, a blocked reply) only drop replies that cannot have steered messages after them. So the deleted restate_agent_steers rows and the companion's taken marks never disagree in practice.

^{｜ Fix all ➔ ｜ Fix 👍s ➔ ｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

pullfrog

ℹ️ I found no defects in the core design. Two framework paths were checked less deeply than the rest (see below), and CI is still pending.

Reviewed changes

I reviewed all 36 files at 3104124: the core steering mechanism, the four framework integrations, the summary and transcript changes, the docs, and the tests and mutants.

steer view and private steered handler (_object.py): the view sends the message to the companion. The companion always queues a steered placeholder for it (idempotency key steer:<id>), so the message keeps its place in the key's queue. The placeholder does nothing if a turn already stored the message.
Companion inbox (_signals.py): the inbox holds messages only for a turn that started with steers: true, at most STEERS = 16 per turn, counting messages already taken. publish now returns the messages still waiting, taken removes their text, and _closed and reset clear the inbox.
take_steers and Turn.steer (_steering.py, _conversation.py): these take messages only at a step with no open calls, while the turn is still going. The store is shielded and runs under the ledger lock. The DB uniqueness check and the step and steer rows commit together with the outbox taken send. A new appended table, restate_agent_steers, has a unique id.
History and views: latest_ask skips steered messages. history(), the budget (_turn_size), summaries and the summary view now include each earlier turn's steered messages. drop deletes steered messages after the dropped reply.
Frameworks:
- OpenAI: the input filter takes messages at each step, and _Steers/_spliced put them back into every later request.
- Pydantic AI: _Steering calls enqueue('asap'). Feedback is now deduplicated per RetryFeedbackPart.
- ADK: the before_model_callback plugin takes messages, for an LlmAgent root only.
- LangGraph: the built-in model node takes them, using steer:<id> ids, and a trimmed request never starts at a steered message.

ℹ️ What I checked myself, and what I checked less deeply

I traced the exactly-once path through each of these cases and found no case where a message is answered twice or never:

the turn crashes after the store (held, plus the unique DB read on retry);
the turn is stopped while waiting for the ledger (going() is checked again under the lock);
the turn is killed (the next start closes it and the placeholder answers the message);
the turn closes before taken arrives (taken is a no-op);
the view runs again (the companion call rejoins through its idempotency key);
the cap is reached;
drop runs after a take (the message is either taken again under the same taken:<id> key or answered by the placeholder).

final_reply() also returns None once a steered message comes last, so a retried turn asks the model again instead of returning a stale answer.

I checked these less deeply: whether _spliced places messages correctly in OpenAI requests after handoffs and user call_model_input_filters, and whether Pydantic AI's enqueue and ADK's plugin ordering behave as the code assumes in the installed versions. I confirmed that RunContext.enqueue exists in the installed Pydantic AI. Beyond that, these rely on the PR's own framework tests.

Local runs:

tests/agents_steering.py: 11 passed.
In-process tests matching -k steer across the framework test files: 13 passed.
agents_conversation, agents_messages and agents_snapshots: 57 passed.
The framework files, run in-process: no failures. Their Docker-harness tests could not run here.
ruff and pyright on the core files: clean.

The CI Lint and Test and integration jobs were still pending when I reviewed.

^{｜ Fix it ➔ ｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

pullfrog · 2026-11-09T11:21:08Z

Follow-up to my review: I've now finished checking how OpenAI places steered messages in later requests (_Steers / _spliced). It holds for parallel tool calls, multi-step runs, handoffs with and without an input_filter, a user's own filter, compaction, _resume and output-guardrail re-asks. One narrow case does not.

A nested Agent.as_tool run gets the outer run's steered messages. When an agent is used as a tool, the SDK passes the outer run_config to the nested Runner.run if none is given (agents/agent.py: resolved_run_config = context.run_config). The nested run's requests then also go through _Hooks.model_input. Once the outer run has taken a message, self.steers.placed(items) (python/restate/agents/openai/_agent.py:329) adds it to every request of the sub-agent, for example [sub prompt, S1, fc z, fo z]. The request is still valid, but the sub-agent sees user messages from the outer conversation, and its item keys are added to self.steers.seen.

Taking messages is not affected: take_steers returns [] while the outer as_tool call is open.

A possible fix: only splice into requests whose input contains this run's own anchor or first item, or whose data.agent belongs to this run. A test with an as_tool sub-agent would pin it.

The other open question from my review, how Pydantic AI, ADK and LangGraph behave in their installed versions, also checked out with no defects.

^{｜ View workflow run ｜ via Pullfrog ｜ Using claude-opus-5-5 ｜ 𝕏}

igalshilman · 2026-11-09T11:25:21Z

Thanks, the as_tool case was real, and it went one step further: the nested run's tool outputs also went through new_results, so the turn stored them without their calls. That breaks the next turn's request.

Fixed in 19a42c5. The run's hooks now count its running tool calls. The SDK makes the run's own next request only once all of its calls have ended, so a request made while one of them runs belongs to a nested run. Such a request gets the user's own filter only: nothing is stored, and no steered message is taken or placed. The fix comes with the test test_an_agent_used_as_a_tool_gets_none_of_the_turns_messages_and_stores_none_of_its_items and the mutant openai-nested-run-filtered, which the test catches.

Generated by Claude Code

pullfrog

✅ No new issues found. The new commit fixes the Agent.as_tool defect raised in the follow-up comment.

Reviewed changes

I reviewed the one commit since 3104124, which keeps a nested OpenAI run (an agent used as a tool) separate from the turn.

_Hooks.running gate: on_tool_start and on_tool_end count the run's open tool calls. While any call is open, model_input applies only the user's own call_model_input_filter (_filtered). A nested run therefore gets none of the turn's steered messages, and the turn stores none of the nested run's tool outputs. I checked this against the installed agents SDK. as_tool passes on run_config (and our filter with it), but uses its own hooks argument rather than the outer run's, so the nested run's on_llm_end and on_tool_* never reach _Hooks. A function tool's on_tool_end follows every on_tool_start except in two cases: a nested interruption, which fails the turn here (APPROVALS), and an error the tool raises, which fails the run. So the counter cannot be left non-zero for a later step.
_filtered extraction: this moved the user-filter call into a helper. The outer path is unchanged.
Test, mutant and docs: added test_an_agent_used_as_a_tool_gets_none_of_the_turns_messages_and_stores_none_of_its_items, which passes locally, and the openai-nested-run-filtered mutant. The module docstring and frameworks.md now describe the nested run's behaviour.

One behaviour change, which I consider acceptable: a nested run's request no longer sets requesting. If a nested model error propagates out of the tool, _go now classifies it as a tool's error (from_tool) rather than a provider refusal. That matches where the error happened from the turn's point of view.

^{｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

… step A key answers one message at a time: a message sent while a turn runs waits in Restate's queue and becomes the next turn, and to change course a client stops the turn and sends again. This adds a second way in, which leaves `chat` as it is: the agent's new `steer` view. The turn being answered takes a steered message at its next step (once the model's tool calls have their results, before its next request) and the model's next request has it; a message that comes after the turn's last step is answered next, as a turn of its own, in the order it was sent. Delivery, in the companion (`signals`) - `steer({id, text})` sends the agent's new private `steered` handler the message at once (idempotency key `steer:<id>`): the message takes its place in the key's queue as it is sent, so it is answered in order with the chats sent after it, and even if the turn dies. - If the turn being answered takes steered messages (`Started.steers`, a field the turn's `start` adds, like `settles`), the companion also holds the message in its `inbox`, up to 16 per turn (`STEERS`). - `publish`, which a turn calls at each tool call's start and end, returns that turn's inbox (`steering`, absent when empty). So a turn learns of a message at no cost of its own: nobody steering costs nothing. - `taken(ids)` forgets messages a turn stored; `_closed` forgets those a closed turn left; `reset` (clear) forgets them all. `status` shows the inbox (`steering`). In the turn - `take_steers(ctx, conversation)` (in `restate.agents.internal`), called by an integration at a step, stores the messages the last `publish` listed that are not stored, as messages of the user that are steps of the turn, and returns them for the model's next request. It does nothing outside a `chat` turn, for an agent whose framework does not say it takes them (`chat(..., steers=True)`), for a background agent, while a tool call has no result, or once the turn stops. - `Turn.steer` stores them in one transaction, each once by its id (in the new table `restate_agent_steers`, an appended migration), with the companion's `taken` sent from the transaction's outbox: once, with the commit. The store is shielded from a stop cancelling it mid-commit. - `steered` answers the message as a turn of its own, unless the database has it (a turn took it): so each message is answered once. A dropped reply drops the messages steered after it, which `steered` then answers. - `latest_ask` skips steered messages: the output check still asks again, ADK still resumes the prompt's invocation, and a graph keeps its run. - The model keeps seeing them: an earlier turn is its prompt, the messages steered to it, and its answer, in `history`, the budget and the summaries. The summary view shows them under their ids. Also `AgentClient.steer` and the terminal chat's `/steer`. Tests: the core in process (stored once, through a crash, after a drop, with the output check, in the history and the summary view, the migration), the companion on a server (the inbox, the placeholder, the cap, a closed turn, clear), and an agent with a scripted framework on a server (a message steered during a tool call reaches the next request; one after the last step, or to an agent that takes none, or with no turn, is answered as a turn of its own in order; a turn killed before its step leaves it to its own turn). 569 passed; the 3 failures are the usual host.docker.internal ones. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

…uard it Docs (the reference pages) - turns.md: a Steering section (the steer view, the step, where a message waits, how the turn learns of it from publish, taken once, answered as a turn of its own otherwise, a taken message is its turn's, what the model and clients see, limits), the words "step" and "steered message", the status and summary views, the revisions, Started.steers alongside settles, and the guarantees: a turn is joined only by a steered message, at a step, and each is answered once. - conversation.md: restate_agent_steers (the first appended migration), a steered message as a step, earlier turns as prompt, steered messages and answer in history, budgets and summaries. - README.md: the bullet, the view, the module. - learnings.md: D6 to D10, why a view of its own, the placeholder queued as the message is sent, the inbox on publish, in place not a restart, and a late message as the next turn. - porting.md: the TypeScript step; access-patterns.md: the traffic. Traffic, measured (scripts/steering_traffic.py, against a real server, any checkout): a turn of one tool call that nobody steers costs what it costs before steering, 47 journal entries and 5 companion calls (start, three publishes, close), on this branch and on main alike. A turn that takes a steered message: 57 entries and 6 calls, the sixth the outbox's taken. The message itself: the steer view (5 entries), the companion's steer (13), and the placeholder that finds it taken (3). Mutants (scripts/resurrect.py, durability, from no fix): the placeholder answering a taken message, latest_ask counting steered messages, history dropping them, a message taken while a call is open, publish without the inbox, taken never sent, a drop keeping their rows, and the summary view hiding them. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

The in-process tier gains agents_steering, the server tier agents_steering_server. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

…step, in place The built-in LangGraph agent (`langgraph.agent(model=...)`) now takes the messages steered to the turn being answered (the `steer` view) at the turn's next step, and says so: `chat(..., steers=True)`. At the step - The model node (`call_model`) calls `take_steers` before its request. The graph reaches that node after the tools node, which stores every tool call's result and then publishes (`tool_call_ended`): the publish's response is how the turn learns of the waiting messages. `take_steers` stores each once, as a message of the user that is a step of the turn, and returns the new ones; it returns nothing, at no cost, when nothing waits, a call has no result, the turn stops, or it is no `chat` turn (a handler of your own that calls `langgraph.prompt`). At the run's start the node calls it too: a message the turn already knows of then goes into the first request. - The request gets them after what the state holds, so after the tool results, and the node returns them with the reply (`{"messages": [*steered, reply]}`), in the order the conversation stored them: the graph's state keeps them for the rounds after. Why in place - Running the graph again with the message would reset LangGraph's recursion limit, which is what ends a turn whose model keeps calling tools. Returned in the model node's update, the messages are appended by the add_messages reducer and the run goes on, within its limit. Marked as steered - In the graph, a steered message has the id `steer:<its id>`, which no other message has. The conversation stores it without one, so `_history` marks the stored ones that the history holds (a retried attempt's own, and earlier turns'), by the entries whose items `history()` returns (`Turn.steered()`, and the transcript's `steers`). - `_latest` trims each request to the model's window with LangChain's `trim_messages(start_on="human")`, and sends the latest prompt whole when nothing fits. A steered message is a `HumanMessage`, so either could have started the request at it, without the turn's prompt and the tool calls and results before it. Now a request starts at the first prompt among the messages that fit, never at a steered message, else at the latest prompt, whole. Without steered messages, requests are as before. `graph=` agents take none - A graph of your own takes no steered messages in this version: its steps are its own, and `_graph_agent` does not pass `steers=True`. So `steer` answers that no turn takes the message, and it is answered as a turn of its own, after the one being answered, in the order it was sent. Docs: the steering point in `_agent.py`'s docstring and `agent()`'s, the `graph=` notes in `_graph.py` and the package's, and LangGraph's section of docs/agents/frameworks.md. Tests, in tests/agents_langgraph.py, on a server: an agent whose scripted model records each request and answers the latest message of the user, with a window that holds no message, so that each request is the latest prompt and what followed it (a steered message taken for a prompt would show). - A message steered while a tool waits on a gate is in the very next request, once, after the tool's result, under `steer:<id>`. The turn's answer is the reply to it; it is stored once, and the summary view shows the prompt, the message and the answer. - A message steered after the first tool round, which has the model call a tool again, is still in the request after the second round, in place. - A crash after the message is stored (the model's request with it fails the attempt): the retried attempt's request is the crashed one, with the message once and marked, and it is not stored again. - `_latest` (in process): a request starts at the turn's prompt, never at a steered message, when only that fits and when nothing does, nor at one of an earlier turn. - tests/agents_langgraph_graph.py: a `graph=` agent's `steer` answers that no turn takes the message, which is answered next, as a turn of its own. Each of these mutations fails them: no state update, no marks on the history, no steered messages in the request, no `steers=True`, `_latest` ignoring the marks. tests/agents_langgraph*.py, agents_steering.py and agents_steering_server.py: 110 passed. All agents tests: 393 passed, and the usual host.docker.internal failure of examples_agents.py. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

…uest An ADK agent whose root is an LlmAgent (the built-in agent, or an agent or App of your own with one) now says it takes steered messages (`chat(..., steers=True)`): a message sent to `steer` while a turn runs joins the turn at its next step, and the model's next request has it. The run goes on in place; it is not restarted. At the step - `_Restate`, this package's plugin, takes the messages in its `before_model_callback`, as a model request goes out that ADK built from the session's events. `take_steers` stores those the last `publish` listed, only once each of the model's tool calls has a result, and each once. Each is appended to that request's contents and, through `_Sessions.steered`, to the in-memory session after its events (not stored again), so that every later request of the run has it, in place. - The plugin is now built per run, with that run's `_Sessions` (it was built with the app, which no longer takes `ctx`): it reaches the conversation without a private accessor of ADK's. It still comes after the app's own plugins. Why at the request, rather than as the tools' results are stored - Taking them in `_Sessions.append_event`, after a function response, was the first plan. ADK makes no request after some function responses: a tool's `skip_summarization` (its result is the answer), an output schema answered through `set_model_response` (ADK makes the final event itself), and, from ADK's code, an auth or confirmation request that ends the invocation. In process, a message steered while such a tool ran was stored by the turn and never shown to the model. At the request, a step that ADK ends takes nothing, and the core answers the message next, as a turn of its own. An invocation of its own - A steered message is an event of the user that starts an invocation of its own (`_Items.prompt`, which the core makes it with), never the running one. After a crash, ADK resumes the prompt's invocation from that invocation's events (`latest_ask` skips steered messages, so `_final` and `_run` keep resuming the prompt's, or the output re-ask's). ADK's classic runner, which runs a root that is no LlmAgent and resumes its LlmAgents from their agent states, takes a message of the user last among those events for the answer to a tool call, and raises `ValueError: No agent to transfer to for resuming agent from function response step` on every retry: with the running invocation's id, a crash after the store fails the turn for good. On ADK 2.14 an LlmAgent root runs on the node runner, which loads no agent states, so a same-id message would resume there; a test that forces a SequentialAgent root to take steered messages shows the failure. - With the default `include_contents`, ADK puts a user event of another invocation in the request (`_should_include_event_in_context` does not filter by invocation): it is in every request after it, in its place. Only LlmAgent roots, for now - `_steers(own)`: the built-in agent, or an agent or App whose root is an LlmAgent. A workflow root (a SequentialAgent, ParallelAgent or LoopAgent, an App's Workflow) passes `steers=False`: each message steered to it is answered as a turn of its own. Limits, documented in the module's docstring - A message sent during the model's last reply, or while a tool runs after whose result ADK asks the model nothing more, is answered next, as a turn of its own. - An agent with `include_contents='none'` starts each request at the latest message of the user: the requests after the one that takes a steered message start at it, without the prompt and the tools' results before it. - An AgentTool is one call: ADK gives its runner our plugin, but the parent's call is open while its agent runs, so its requests take nothing; the step is after its result. - An agent's own before_model_callback runs after the plugins: one that answers in the model's place gets the request with the message in it. Tests, in tests/agents_adk.py (against a Restate server unless in process) - A message steered while a tool runs reaches the next request, after the function response, once; the model answers it with a call of its own and the request after that still has it in place; it is stored once (its own queued turn does nothing), and the summary view shows prompt, steered message, answer. - A crash on the request with the message: the retry resumes the prompt's invocation, its request has the message once, the tool ran once, the model was not asked for the call again, and nothing is stored twice. - A SequentialAgent root takes no steered message (`turn` is None), which is answered as the next turn. - A message steered during the last model call is answered as the next turn. - In process: after a tool's skip_summarization, and after an output schema's set_model_response, no request follows and the turn stores nothing. - In process: on ADK's classic runner (a SequentialAgent root made to take steered messages), the retry after a crash resumes past the message. Each fails without what it covers (taking at the function response, steers for no root or for every root, no append to the request, no append to the session, the running invocation's id). tests/agents_adk.py 34 passed (27 before), tests/agents_steering.py 9, tests/agents_steering_server.py 5; ruff, pyright (0 errors) and mypy clean. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

The Pydantic AI agent now says it takes steered messages (`steers=True` in `agent()`'s `chat`), and every run takes them at each step, with a capability of its own (`_Steering`). At the boundary - Before each model request, once the model's tool calls all have their results stored (by `_Restate`, earlier in the same before-chain), it calls `take_steers`: the messages that wait are stored once, as messages of the user that are steps of the turn (`ModelRequest(parts=[UserPromptPart])`), and returned. - It then enqueues those same stored requests with Pydantic AI's own `RunContext.enqueue(..., priority="asap")`. Content enqueued from a `before_model_request` hook is delivered once the whole before-chain ran (`_agent_graph._apply_before_model_request`), into the request being prepared and into the run's history (`ctx.messages`), after what both hold: right after the tools' results. So what the model gets is what is stored, where it is stored, and a retried attempt or the next turn rebuilds the same history from the database. - In place: the run carries on with the message. Starting `agent.run` again from the stored history would reset the run's usage limits, its retries and its capabilities' state. - A whole `ModelRequest` enqueued outside a tool call is never spliced into the request that carries the tools' results (Pydantic AI splices only parts enqueued during a tool call, and `ctx.tool_call_id` is None in `before_model_request`): it is a request of its own, which `_Restate` never stores again (it holds no result and no feedback). - It takes nothing for a resumed suspended response: that request ends with the paused `ModelResponse`, into which Pydantic AI delivers nothing, so a message taken then would reach the model after the response though stored before it. The next step takes it. - Outside a `chat` turn that takes steered messages (a `prompt` in a handler of one's own, a background agent), `take_steers` returns nothing, at no cost. Snapshots - `_Steering` is innermost, ordered after `_Snapshots` (`CapabilityOrdering(position="innermost", wrapped_by=[_Snapshots])`). A snapshot covers every position stored so far (`ledger.next - 1`) and holds the request's messages as the before-chain left them, before the delivery: a message stored before a snapshot of the same request would be covered by it but not in it, and lost from the next turn's history and from a retried attempt. Stored after it, it comes after the snapshot's items. - Without snapshots (a `chat` of one's own that answers with `prompt`), the ordering has nothing to follow, and the turn projection has the messages in their turn. Feedback found by its parts (`_Restate._new`) - A run merges the consecutive requests it is given, and resumes from the last one as a new `ModelRequest` with the same parts. `_Restate` found Pydantic AI's feedback (`RetryFeedbackPart`) by the whole request's JSON, so a merged or resumed feedback request was taken as new and stored again, then again in every later turn. That was so before steering: a retry after a stored feedback stored it twice. Steering makes consecutive requests common (a result, or a feedback, then the steered message): after a retry that took a message right after a stored feedback, the next turn stored the feedback and the message again, and the views showed the message twice. Each `RetryFeedbackPart` is now found by its own JSON (it has its own time), which merges and resumes keep. Limitations - A turn learns of a message only from the companion's `publish`, which a tool call's start and end make: a step that ran no tool (Pydantic AI's feedback on an answer that does not validate, say) takes only what the last publish listed. A message that comes during the last model request is answered next, as a turn of its own. - Not changed here: what other capabilities enqueue as requests of their own (the harness's Planning, SystemReminders, DetectLoops, WarnNearLimits, Memory, ...) reaches the model and the run's history, but `_Restate` stores only results and feedback, so a retried attempt and the next turn do not have it. Tests - On a server, with a scripted model and a tool held on a `support.Gate`: a message steered during the tool call reaches the very next request, after the tool's result, once; the turn's answer is the reply to it; it is stored once, the summary view shows the prompt, the message and the answer, and its queued turn of its own does nothing. A crash while the model answers it: the retried request has it once, and it is not stored again. A message steered during the last model request is answered as the next turn. - In process: a compaction (of one's own, at the step) and a steered message in the same request: the snapshot does not hold it, and a retried attempt and the next turn get it after the snapshot's items, once (this fails with `_Steering` ordered before `_Snapshots`). A `chat` of one's own that answers with `prompt`, without snapshots, takes two messages, in order. A feedback, a crash, a retry that takes a message right after the feedback, and the next turn: each stored once, and given to the model once (this fails with the former `_new`). - tests/agents_pydantic_ai.py, agents_pydantic_ai_snapshots.py, agents_pydantic_ai_stores.py, agents_steering.py and agents_steering_server.py: 84 passed (78 before). ruff, pyright and mypy are clean. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

An OpenAI agent's turn now takes the messages steered to it (the agent's steer view) at its next step, and its run goes on with them: agent() says so to chat (steers=True), and the run's call_model_input_filter, which already stores the tools' outputs right before each model request, takes them there. At a step - The input filter (_Hooks.model_input) stores the new tool outputs, then calls take_steers: once each of the model's tool calls has its output, the turn stores the waiting messages, once each, as messages of the user, and the request about to go out has them right after those outputs. - A run also takes the messages waiting as it starts (_go, after _resume closed a cut-off reply): they are stored before Runner.run, so they are in its input. A request whose last item cannot be told apart in later requests (a message of the user, as a run's first request ends with) takes none: its messages wait for the next step. In place, not a restart - A new Runner.run from the stored history would run the input guardrails again, start max_turns over, and reset the tool-choice tracking, the agent-start hooks and the trace. The filter changes only the request about to go out, and the run goes on as it was. The splice - The SDK makes each request's input afresh (the run's input plus its generated items, copied for the filter), so a message put in one request is gone from the next. The run keeps the messages it took, each with the key of the item it came after (a tool call's or output's call_id, else the item's own id: _key), and puts each back in every later request right after that item. Messages taken together keep their order; an item found twice counts where its last copy is, the one the SDK's dedup keeps. User messages have no id, so that dedup never drops them. - If a handoff's input_filter took that item out, the message goes right before the first item that came after it (one whose key the run first saw after taking it), as if the filter had kept the message; else at the end. - The run's own call_model_input_filter (a ToolOutputTrimmer, say) still runs after, on the input with the messages. On a retry - Stored messages come back in conversation.history() at their places, after the outputs they followed, so a resumed run's input has each once; take_steers never returns one twice, and the new run splices only what it took itself. A stored message after the outputs leaves _resume no open call, and makes run() ask the model rather than take a final reply. Limitations - A message is taken only at a step: one that comes while the model writes its final answer is answered next, as a turn of its own (the agent's steered handler); one sent during a long tool call waits for it. - A handoff whose input_filter takes its own output out of the next agent's input (handoff_filters.remove_all_tools does) leaves the handoff call without an output in the turn, so take_steers takes nothing for the rest of that run (by design: the turn has an open call). - The fallback placement counts an item with no key (a Chat Completions message, whose id is the SDK's fake one) as having come before the message. - The SDK's ToolOutputTrimmer counts user messages as turns, so a steered message makes the outputs before it older to it, as on a resumed run. Found, not changed (independent of steering): a handoff whose input_filter removes the handoff's own output never has that output stored, so every later request of the conversation carries the transfer call without an output, which OpenAI's APIs reject. The same on the base commit. Tests (tests/agents_openai.py) - Against a server: a message steered while a tool waits on a gate is in the next request, once, right after the tool's output; the turn's answer is the reply to it; the summary view shows the prompt, the message and the answer; its queued turn of its own finds it taken. A message steered while the model writes its final answer is the next turn. - In process, with a companion whose publish lists the messages: a message taken after the first of two tool rounds is in the request after the second, in its place, once, stored once and taken once; after a crash once the filter stored it, the retry's request has it once and it is not stored again; an input guardrail runs once; a message waiting as the run starts is in its first request; after a remove_all_tools handoff the message keeps its place and the run takes no more; the splice itself (order, a repeated item, the fallback). - A mutation of each fails a test: no steers=True, the current request only, no take as the run starts, always at the end, the fallback at the end, the first copy of an item, a restart instead of in place (the guardrail test). tests/agents_openai.py 54 passed (46 + 8), agents_steering.py 9 and agents_steering_server.py 5; the whole suite 577 passed, with the 3 usual host.docker.internal failures. ruff, pyright and mypy are clean. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

frameworks.md gains a row for steered messages and a note per framework: OpenAI splices them back into each later request, Pydantic AI enqueues them after any snapshot, ADK adds them as a request goes out, for an LlmAgent root. It also fixes the note on which capability is innermost, and records two gaps found on the way: an OpenAI handoff filter that drops the handoff's output, and what other Pydantic AI capabilities enqueue, which is not stored. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

…e fixes of a self review - The cap of 16 counted only the messages a turn had not taken yet, so a turn steered on and on took them without end. `taken` now marks a message (without its text) rather than forgetting it, until the turn ends: the cap counts them all, and a message sent again is still the turn's. - `taken` publishes only a change: one for a turn already closed wakes no follower. - A turn that stopped while it waited for the ledger to store a message takes none, which is then answered as a turn of its own. A message the turn did take shares its fate (stopped, refused, failed), as documented. - The turn just answered counts its steered messages toward the budget in the same attempt (`prompt.steers`), so a summary due is asked for now rather than a turn later; a drop trims them. - `publish` reads the companion's turn only when a message waits, and `status` reads it once; `_steps` reads the steered marks with one join; `Turn.steer` uses `_INSERT_STEP`. - ADK: the agent's own before_model_callback runs after the plugin's, on the request with the message, so a response it gives answers it too. Three more mutants in scripts/resurrect.py guard the cap, the stop and the budget; the publish mutant follows the code it mutates. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

A turn is now its prompt, the messages steered to it that it took, and what answers them: turns.md says so where it defines a turn, its life (a step takes the messages its publish lists), how it ends (the messages it took go as it goes, those it did not are answered after it), its attempts, stopping and clear, the turn model and the guarantees. "One message at a time" is now "One turn at a time". Steering a turn opens with what happens to a steered message, then three cases as sequence diagrams: taken at the next step, too late for the turn, and a turn that stops. README and conversation.md follow. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

…s items The conversation now makes and stores a steered message as the framework's own item, so no integration works out which messages were steered: - `Items.steered(text, steer_id)`, optional like `summary`, makes the item of a steered message (by default a prompt's). LangGraph's carries the id `steer:<id>` and is stored with it, so its history, a retried attempt's and the next turns', tells steered messages from prompts with no pass that marks them again (`_history` and `_as_steered` are gone). ADK's says why it starts an invocation of its own. - `take_steers` returns the items the request gets, not entries: no integration reads an `Entry` for them any more. - `Turn.steer` skips the messages the turn already holds, and takes `taken` and `going` as required callbacks. Its stored-id query is shared with `steer_stored`. Readability: - ADK's plugin takes the messages at its step. `_Sessions` only adds the stored events to the session (`add_stored`), and no longer needs `ctx`. - OpenAI's in-run state for the messages a run took, which each later request gets again, is one small class (`_Steers`), out of the run hooks. - The turn's copy of its waiting messages is `Inbox` (it was `Steering`, a name `_api` also gives another type), and the companion's `steer` takes `_api.Steer`, which `SteerRequest` duplicated. - One `_turn_size` for the budget and the transcript's walk, and one helper for the summary view's messages of the user. Mutants: the steering ones follow the code, and `steer-langgraph-unmarked` is new. `langgraph-reask-stored-reply` matched no code since the steering change, and now does. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

…st a server agents_object.py has no steering tests: the companion's are in agents_signals.py, and the agent's steered placeholder's in agents_steering_server.py (Pullfrog's review). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

`Agent.as_tool` runs its agent with the outer run's `run_config` when given none, so its requests went through the turn's input filter: they got the messages steered to the turn, and the turn stored the nested run's tool outputs, without their calls (which breaks the next turn's request). A request made while one of the run's tool calls runs is a nested run's (the SDK makes the run's own next request once all of its calls end): it gets the user's own filter only. Test and mutant `openai-nested-run-filtered`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

igalshilman marked this pull request as ready for review November 9, 2026 10:24

pullfrog Bot reviewed Nov 9, 2026

View reviewed changes

Comment thread tests/agents_steering.py Resolved

pullfrog Bot reviewed Nov 9, 2026

View reviewed changes

pullfrog Bot approved these changes Nov 9, 2026

View reviewed changes

claude added 13 commits November 9, 2026 14:20

Agents: testing.md lists the steering tests

c0c7c36

The in-process tier gains agents_steering, the server tier agents_steering_server. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

igalshilman force-pushed the claude/agents-steering branch from 19a42c5 to fe79371 Compare November 9, 2026 14:31

igalshilman merged commit ecf01f0 into restatedev:main Nov 9, 2026
11 checks passed

github-actions Bot locked and limited conversation to collaborators Nov 9, 2026

Navigation Menu

Repository navigation

Agents: steering, a message joins the turn being answered at its next step - #301

Agents: steering, a message joins the turn being answered at its next step#301
igalshilman merged 13 commits into
restatedev:mainfrom
igalshilman:claude/agents-steering

igalshilman commented Nov 9, 2026 •

edited

Loading

Uh oh!

igalshilman commented Nov 9, 2026

Uh oh!

pullfrog Bot left a comment

Uh oh!

Uh oh!

pullfrog Bot left a comment

Uh oh!

pullfrog Bot commented Nov 9, 2026

Uh oh!

igalshilman commented Nov 9, 2026

Uh oh!

pullfrog Bot left a comment

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

igalshilman commented Nov 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Turn semantics, before and after

What happens to a steered message

How it works

Taken at the next step

Too late: the next turn

The turn is stopped

A crash after the message is stored

What the model and clients see

Frameworks

Limits

Traffic

Compatibility

Testing

Found on the way

Docs

Uh oh!

igalshilman commented Nov 9, 2026

Uh oh!

pullfrog Bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

pullfrog Bot left a comment

Choose a reason for hiding this comment

ℹ️ What I checked myself, and what I checked less deeply

Uh oh!

pullfrog Bot commented Nov 9, 2026

Uh oh!

igalshilman commented Nov 9, 2026

Uh oh!

pullfrog Bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

igalshilman commented Nov 9, 2026 •

edited

Loading